// src/main/java/mqtt/dromara/org/controller/MqttController.java
package org.dromara.mqtt.controller;

import cn.dev33.satoken.annotation.SaIgnore;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.dromara.mqtt.service.MqttMessageHandler;
import org.dromara.mqtt.service.impl.MqttClientService;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.web.bind.annotation.*;

import java.util.concurrent.ConcurrentHashMap;

@RestController
@RequestMapping("/api/mqtt")
@Slf4j
public class MqttController {

    @Resource
    private MqttClientService mqttClientService;

    @Resource
    private MqttMessageHandler mqttMessageHandler;

    private final ConcurrentHashMap<String, IMqttMessageListener> topicListeners = new ConcurrentHashMap<>();

    /**
     * 发布消息
     */
    @PostMapping("/publish")
    public String publishMessage(@RequestParam String topic,
                               @RequestParam String message,
                               @RequestParam(defaultValue = "1") int qos,
                               @RequestParam(defaultValue = "false") boolean retained) {
        try {
            mqttClientService.publish(topic, message, qos, retained);
            return "消息发布成功";
        } catch (MqttException e) {
            log.error("发布消息失败", e);
            return "消息发布失败: " + e.getMessage();
        }
    }

    /**
     * 订阅主题
     */
    @SaIgnore
    @PostMapping("/subscribe")
    public String subscribeTopic(@RequestParam String topic,
                               @RequestParam(defaultValue = "1") int qos) {
        try {
            // 创建消息监听器
            IMqttMessageListener listener = (t, msg) -> {
                String payload = new String(msg.getPayload());
                log.info("收到消息: topic={}, message={}", t, payload);

                // 使用注册的消息处理器处理消息
                if (mqttMessageHandler != null) {
                    mqttMessageHandler.handleMessage(t, payload);
                }
            };

            mqttClientService.subscribe(topic, qos, listener);
            topicListeners.put(topic, listener);
            return "订阅成功";
        } catch (MqttException e) {
            log.error("订阅失败", e);
            return "订阅失败: " + e.getMessage();
        }
    }

    /**
     * 订阅主题并使用默认消息处理器
     */
    @SaIgnore
    @PostMapping("/subscribeWithHandler")
    public String subscribeWithHandler(@RequestParam String topic,
                                     @RequestParam(defaultValue = "1") int qos) {
        try {
            mqttClientService.subscribeWithHandler(topic, qos);
            return "订阅成功";
        } catch (MqttException e) {
            log.error("订阅失败", e);
            return "订阅失败: " + e.getMessage();
        }
    }

    /**
     * 取消订阅
     */
    @PostMapping("/unsubscribe")
    public String unsubscribeTopic(@RequestParam String topic) {
        try {
            mqttClientService.unsubscribe(topic);
            topicListeners.remove(topic);
            return "取消订阅成功";
        } catch (MqttException e) {
            log.error("取消订阅失败", e);
            return "取消订阅失败: " + e.getMessage();
        }
    }

    /**
     * 检查连接状态
     */
    @GetMapping("/status")
    public String getConnectionStatus() {
        return mqttClientService.isConnected() ? "已连接" : "未连接";
    }
}
